airbyte_cdk/sources/declarative/decoders/__init__.py (1)

10-10: Nice import of CompositeRawDecoder.
This addition expands the module's offerings. Perhaps we might include a docstring in the composite_decoder.py file to explain its use cases for future readers, wdyt?

airbyte_cdk/sources/declarative/decoders/composite_decoder.py (4)

18-19: Interface design looks clear.
Introducing Parser as an ABC is a neat approach to unify custom parsers. Perhaps we could add a short docstring on the intended usage or common parameters, so new contributors can adopt the pattern seamlessly, wdyt?

---

55-58: Potential optimization for error handling.
When JSON decoding fails, we log a warning and skip. This is good for resilience, but might it be helpful to have a configurable option to raise an exception if unexpected lines are encountered, wdyt?

---

76-83: Potential multi-chunk reading concern.
Reading large CSVs in chunks is effective. Could we consider memory usage limits or advanced chunk sizing if extremely large files are processed, wdyt?

---

95-104: CompositeRawDecoder usage clarity.
Implementation looks straightforward, but could we include an inline doc or example clarifying what happens if the underlying parser fails on a line (e.g., how are partial results handled), wdyt?

unit_tests/sources/declarative/decoders/test_decoders_memory_usage.py (2)

25-31: Pragmatic approach for test data generation.
Generating 2 million lines is a hefty load. Should we consider adding an environment-based toggle or a smaller default for local testing, wdyt?

---

35-36: Tip on skipping slow tests.
We decorate this with @pytest.mark.slow. Maybe we can also provide a quick way to skip slow tests locally unless explicitly enabled, wdyt?

unit_tests/sources/declarative/decoders/test_composite_decoder.py (2)

55-67: Gzip CSV decoding test coverage is good!
Testing with multiple encodings ensures robust coverage. Could we also add a test scenario for missing headers or partial CSV lines, wdyt?

---

93-106: Gzip JSON lines test scenario is comprehensive.
We check repeated iteration logic and assert record counts. Might it be useful to test a scenario containing at least one invalid line within a gzipped bundle, wdyt?

airbyte_cdk/sources/declarative/models/declarative_component_schema.py (3)

1112-1114: Consider clarifying the class docstring.
Would you consider adding a short docstring to explain the JsonLineParser's usage? This can help future maintainers easily identify its purpose. wdyt?

---

1116-1121: Check consistency with other parsers.
You might consider adding docstrings or class-level comments like with JsonLineParser for uniformity. wdyt?

---

1536-1540: Allowing flexible parser composition.
Might it be helpful to add a small example or usage note on how the CompositeRawDecoder is configured, especially with nested formats? This could reduce confusion for new developers. wdyt?

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py (5)

146-148: CsvParserModel import.
Nice addition. Could we also provide a quick usage example in the docstring for create_csv_parser to guide developers? wdyt?

---

242-244: JsonLineParserModel import.
Everything looks aligned. Possibly add a note or reference in docstrings to ensure clarity for new maintainers. wdyt?

---

461-461: Adding CompositeRawDecoder constructor mapping.
This is a straightforward approach. Perhaps confirm we log or handle the scenario if a user tries to combine multiple decoders? wdyt?

---

1690-1695: Clean approach for JsonLineParser creation.
Consider adding input validation for cases where encoding is None to avoid runtime errors if a user sets it unexpectedly. wdyt?

---

1720-1725: CompositeRawDecoder creation method.
Perhaps log the selected parser type at runtime to aid debugging. wdyt?

airbyte_cdk/sources/declarative/declarative_component_schema.yaml (3)

2726-2728: Experimental annotation in CompositeRawDecoder.
Would you consider a short note specifying potential changes to the interface, to caution users about relying on it immediately? wdyt?

---

2756-2761: JsonLineParser YAML entry.
Everything here is straightforward. Maybe add a quick usage snippet explaining how the encoding field is applied. wdyt?

---

2762-2774: CsvParser defaults.
Would you consider adding an example for unusual delimiters like "\t" or ";" to the documentation to make it more explicit? wdyt?

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

#!/bin/bash
# Search for CompositeRawDecoderModel usage in factory methods
ast-grep --pattern 'def $_(self, $_, CompositeRawDecoderModel, $_) { $$$ }'

# Search for any references to CompositeRawDecoderModel
rg "CompositeRawDecoderModel" -A 2

# Look for factory-related files that might need to reference this model
fd "factory" --type f

#!/bin/bash
# Check test coverage for CompositeRawDecoderModel
rg "CompositeRawDecoderModel" unit_tests/sources/declarative/parsers/test_model_to_component_factory.py -A 3

# Look for test cases related to composite raw decoder
ast-grep --pattern 'def test_$_composite_raw_decoder$_($$$) { $$$ }' unit_tests/sources/declarative/parsers/test_model_to_component_factory.py

#!/bin/bash
# Search for any composite decoder related tests with a broader pattern
rg "composite.*decoder" -i unit_tests/sources/declarative/parsers/test_model_to_component_factory.py

# Also check the actual implementation details of the factory method
rg "create_composite_raw_decoder" -A 5 airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

#!/bin/bash
# Search for decoder definitions in the schema file
rg -A 5 "CompositeRawDecoder|RawDecoder" airbyte_cdk/sources/declarative/declarative_component_schema.yaml

# Search for any decoder implementations or references
ast-grep --pattern 'class $_ implements RawDecoder'
ast-grep --pattern 'class $_ extends RawDecoder'
ast-grep --pattern 'class CompositeRawDecoder'

# Look for any usage patterns of decoders
rg -A 3 "decoder.*:" airbyte_cdk/sources/declarative/

#!/bin/bash
# Search for parser definitions that might be used with CompositeRawDecoder
rg -A 10 "GzipParser|JsonLineParser|CsvParser" airbyte_cdk/sources/declarative/declarative_component_schema.yaml

# Look for any implementations or usages of these parsers
ast-grep --pattern 'class $_(GzipParser|JsonLineParser|CsvParser)'
rg "type.*: .*Parser" airbyte_cdk/sources/declarative/

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py (3)

1690-1695: Consider handling None encoding?

Would you like to ensure there's a safe default or check when model.encoding is None, so we avoid unexpected decoding errors? wdyt?

---

1718-1721: Handle defaults for CSV parser options?

Would you like to set a fallback if model.delimiter or model.encoding is None, or raise a clearer error? wdyt?

---

1722-1727: Validate parser configuration?

Would you be open to adding a quick check if model.parser is None or invalid? This may prevent unexpected runtime errors. wdyt?

airbyte_cdk/sources/declarative/decoders/composite_decoder.py (3)

18-29: Naming consideration for Parser?

Would you consider something more descriptive like "DataParser"? Or are you comfortable with the generic name? wdyt?

---

47-61: Graceful handling of malformed JSON lines.

You currently skip lines that aren’t valid JSON. If you anticipate a high failure rate, would it be helpful to optionally stop or store such lines? wdyt?

---

63-83: About the TODO note.

There’s a TODO: "migrate implementation to re-use file-base classes." Would you like help creating a GitHub issue or add the changes in this PR? wdyt?

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

airbyte_cdk/sources/declarative/decoders/composite_decoder.py (5)

18-29: How about enhancing the docstring for better clarity? wdyt?

The docstring could provide more details about:

• Expected format/structure of the yielded dictionaries
• Any specific error handling expectations
• Whether the parser should close the input stream

Example:

     def parse(
         self,
         data: BufferedIOBase,
     ) -> Generator[MutableMapping[str, Any], None, None]:
         """
-        Parse data and yield dictionaries.
+        Parse data from a buffered stream and yield dictionaries.
+
+        Args:
+            data: A buffered stream containing the data to parse
+
+        Returns:
+            A generator yielding parsed records as dictionaries
+
+        Raises:
+            ValueError: If the data cannot be parsed according to the expected format
         """
         pass

---

35-43: Should we add error handling for corrupted gzip data? wdyt?

The current implementation might raise uncaught exceptions for corrupted gzip data. Consider wrapping the gzip operations in a try-except block:

     def parse(
         self,
         data: BufferedIOBase,
     ) -> Generator[MutableMapping[str, Any], None, None]:
-        gzipobj = gzip.GzipFile(fileobj=data, mode="rb")
-        yield from self.inner_parser.parse(gzipobj)
+        try:
+            with gzip.GzipFile(fileobj=data, mode="rb") as gzipobj:
+                yield from self.inner_parser.parse(gzipobj)
+        except gzip.BadGzipFile as e:
+            raise ValueError(f"Failed to decompress gzip data: {e}")

---

50-59: How about enhancing the error logging with more context? wdyt?

The current warning message could be more helpful by including line numbers and truncated content:

     def parse(
         self,
         data: BufferedIOBase,
     ) -> Generator[MutableMapping[str, Any], None, None]:
-        for line in data:
+        for line_num, line in enumerate(data, 1):
             try:
                 yield json.loads(line.decode(encoding=self.encoding or "utf-8"))
             except json.JSONDecodeError:
-                logger.warning(f"Cannot decode/parse line {line!r} as JSON")
+                decoded_line = line.decode(encoding=self.encoding or "utf-8")
+                preview = decoded_line[:100] + "..." if len(decoded_line) > 100 else decoded_line
+                logger.warning(
+                    f"Cannot parse line {line_num} as JSON. Preview: {preview!r}"
+                )

---

67-80: Consider using csv.DictReader instead of pandas for simpler CSV parsing? wdyt?

Using pandas for basic CSV parsing might be overkill, especially since we're only using basic features. The standard library's csv.DictReader could be more lightweight:

     def parse(
         self,
         data: BufferedIOBase,
     ) -> Generator[MutableMapping[str, Any], None, None]:
-        reader = pd.read_csv(  # type: ignore
-            data, sep=self.delimiter, iterator=True, dtype=object, encoding=self.encoding
-        )
-        for chunk in reader:
-            chunk = chunk.replace({nan: None}).to_dict(orient="records")
-            for row in chunk:
-                yield row
+        text_data = data.read().decode(encoding=self.encoding or "utf-8")
+        csv_reader = csv.DictReader(
+            text_data.splitlines(),
+            delimiter=self.delimiter
+        )
+        yield from csv_reader

This would:

• Remove the pandas dependency
• Simplify the code
• Handle the None/NaN conversion automatically

---

83-102: Should we improve type safety and documentation for response.raw handling? wdyt?

The # type: ignore[arg-type] suggests a type mismatch between response.raw and BufferedIOBase. Consider:

1. Adding type hints to clarify the expected type:

     def decode(
         self, response: requests.Response
     ) -> Generator[MutableMapping[str, Any], None, None]:
+        # response.raw is a urllib3.response.HTTPResponse which is compatible with BufferedIOBase
         yield from self.parser.parse(data=response.raw)  # type: ignore[arg-type]

2. Enhancing the class docstring with details about response.raw handling:

     """
     Decoder strategy to transform a requests.Response into a Generator[MutableMapping[str, Any], None, None]
     passed response.raw to parser(s).
     Note: response.raw is not decoded/decompressed by default.
     parsers should be instantiated recursively.
+
+    Important:
+    - The response.raw attribute is a raw HTTP response from urllib3
+    - It must be read in binary mode to properly handle compression
+    - The stream is automatically closed when fully consumed
+
     Example:
     composite_decoder = CompositeDecoder(parser=GzipParser(inner_parser=JsonLineParser(encoding="iso-8859-1")))
     """

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

airbyte_cdk/sources/declarative/declarative_component_schema.yaml (2)

2773-2788: Would you consider enhancing the documentation for CompositeRawDecoder?

The experimental component could benefit from:

1. A more detailed description explaining its purpose and use cases
2. Example configurations showing how to chain parsers
3. Clear warnings about potential breaking changes since it's experimental

wdyt?

---

2809-2821: Would you consider adding more CSV configuration options?

The CsvParser could benefit from additional configuration options commonly needed when parsing CSV files:

• quote_char: Character used for quoting fields
• escape_char: Character used for escaping special characters
• has_header: Boolean indicating if the first row is a header
• skip_rows: Number of rows to skip from the beginning

These options would make the parser more flexible for various CSV formats. wdyt?

airbyte_cdk/sources/declarative/models/declarative_component_schema.py (4)

1128-1130: Consider adding encoding validation.

Would it help to validate the encoding parameter against Python's standard encodings to prevent runtime errors? For example, using codecs.lookup() to verify the encoding exists. wdyt?

from codecs import lookup

class JsonLineParser(BaseModel):
    encoding: Optional[str] = "utf-8"

    @validator("encoding")
    def validate_encoding(cls, v):
        try:
            if v is not None:
                lookup(v)
            return v
        except LookupError:
            raise ValueError(f"Unknown encoding: {v}")

---

1132-1135: Consider adding parameter validation.

Would it be helpful to add validation for both parameters? For example:

1. Validate encoding like in JsonLineParser
2. Ensure delimiter is exactly one character long

class CsvParser(BaseModel):
    type: Literal["CsvParser"]
    encoding: Optional[str] = "utf-8"
    delimiter: Optional[str] = ","

    @validator("encoding")
    def validate_encoding(cls, v):
        try:
            if v is not None:
                lookup(v)
            return v
        except LookupError:
            raise ValueError(f"Unknown encoding: {v}")

    @validator("delimiter")
    def validate_delimiter(cls, v):
        if v is not None and len(v) != 1:
            raise ValueError("Delimiter must be exactly one character")
        return v

---

1517-1520: Consider validating inner_parser.

Would it be helpful to ensure inner_parser is always provided since it's required for the composite pattern to work? We could add a validator to raise a descriptive error if it's missing. wdyt?

class GzipParser(BaseModel):
    type: Literal["GzipParser"]
    inner_parser: Union[JsonLineParser, CsvParser]

    @validator("inner_parser")
    def validate_inner_parser(cls, v):
        if v is None:
            raise ValueError("inner_parser is required for GzipParser")
        return v

---

1552-1555: Consider validating parser.

Similar to GzipParser, would it be helpful to validate that parser is always provided? We could add a validator to raise a descriptive error if it's missing. wdyt?

class CompositeRawDecoder(BaseModel):
    type: Literal["CompositeRawDecoder"]
    parser: Union[GzipParser, JsonLineParser, CsvParser]

    @validator("parser")
    def validate_parser(cls, v):
        if v is None:
            raise ValueError("parser is required for CompositeRawDecoder")
        return v

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py (4)

1725-1729: Consider adding error handling.

Would it be helpful to add try-catch block to handle potential encoding errors early? This could provide better error messages during parser creation rather than at runtime. wdyt?

@staticmethod
def create_jsonline_parser(
    model: JsonLineParserModel, config: Config, **kwargs: Any
) -> JsonLineParser:
    try:
        if model.encoding:
            codecs.lookup(model.encoding)
        return JsonLineParser(encoding=model.encoding)
    except LookupError:
        raise ValueError(f"Invalid encoding '{model.encoding}' specified for JsonLineParser")

---

1747-1751: Consider adding error handling for inner parser creation.

Would it be helpful to add try-catch block to handle potential errors during inner parser creation? This could provide better error messages about which part of the composite failed. wdyt?

def create_gzip_parser(
    self, model: GzipParserModel, config: Config, **kwargs: Any
) -> GzipParser:
    try:
        inner_parser = self._create_component_from_model(model=model.inner_parser, config=config)
        return GzipParser(inner_parser=inner_parser)
    except Exception as e:
        raise ValueError(f"Failed to create inner parser for GzipParser: {str(e)}")

---

1754-1755: Consider adding error handling.

Would it be helpful to add validation for both encoding and delimiter during parser creation? This could catch configuration issues early. wdyt?

@staticmethod
def create_csv_parser(model: CsvParserModel, config: Config, **kwargs: Any) -> CsvParser:
    try:
        if model.encoding:
            codecs.lookup(model.encoding)
        if model.delimiter and len(model.delimiter) != 1:
            raise ValueError("Delimiter must be exactly one character")
        return CsvParser(encoding=model.encoding, delimiter=model.delimiter)
    except LookupError:
        raise ValueError(f"Invalid encoding '{model.encoding}' specified for CsvParser")

---

1757-1761: Consider adding error handling for parser creation.

Would it be helpful to add try-catch block to handle potential errors during parser creation? This could provide better error messages about which parser failed to be created. wdyt?

def create_composite_raw_decoder(
    self, model: CompositeRawDecoderModel, config: Config, **kwargs: Any
) -> CompositeRawDecoder:
    try:
        parser = self._create_component_from_model(model=model.parser, config=config)
        return CompositeRawDecoder(parser=parser)
    except Exception as e:
        raise ValueError(f"Failed to create parser for CompositeRawDecoder: {str(e)}")

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

  JsonLineParser:
    type: object
+   required:
+     - type
    properties:
+     type:
+       type: string
+       enum: [ JsonLineParser ]
      encoding:
        type: string
        default: utf-8

#!/bin/bash
# Search for existing decoder implementations that might be affected
echo "Searching for decoder implementations..."
rg -l "class.*Decoder.*\(.*\):" --type python

# Search for decoder usage in existing manifests
echo "Searching for decoder usage in manifests..."
rg -l "decoder:" --type yaml

#!/bin/bash
# Let's try searching without file type restrictions since python extension wasn't recognized
echo "Searching for decoder implementations..."
rg -l "class.*Decoder.*\(.*\):"

# Let's also look for specific decoder usage patterns
echo "Searching for decoder references in code..."
rg "Decoder" -A 3

# Check for any existing composite decoder patterns
echo "Searching for composite decoder patterns..."
rg "CompositeRawDecoder|composite.*decoder" -i